package org.apache.rocketmq.proxy.processor;

import io.netty.channel.Channel;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
import org.apache.rocketmq.broker.client.ProducerChangeListener;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
import org.apache.rocketmq.common.utils.AbstractStartAndShutdown;
import org.apache.rocketmq.proxy.common.Address;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.service.ServiceManager;
import org.apache.rocketmq.proxy.service.ServiceManagerFactory;
import org.apache.rocketmq.proxy.service.metadata.MetadataService;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
import org.apache.rocketmq.proxy.service.route.ProxyTopicRouteData;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;

import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2024/12/30
 * @方法描述：默认消息处理器，这个处理器在第三版本代码中还看不出明显的作用，等到了第四版本代码，大家就知道它发挥作用的原理了
 */
public class DefaultMessagingProcessor extends AbstractStartAndShutdown implements MessagingProcessor {

    private final ThreadPoolExecutor producerProcessorExecutor;
    //服务管理器
    protected ServiceManager serviceManager;

    //第十版本代码新引入的客户端处理器
    protected ClientProcessor clientProcessor;
    //第十二版本代码引入的生产者处理器
    protected ProducerProcessor producerProcessor;
    //构造方法
    protected DefaultMessagingProcessor(ServiceManager serviceManager) {
        ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
        this.producerProcessorExecutor = ThreadPoolMonitor.createAndMonitor(
                proxyConfig.getProducerProcessorThreadPoolNums(),
                proxyConfig.getProducerProcessorThreadPoolNums(),
                1,
                TimeUnit.MINUTES,
                "ProducerProcessorExecutor",
                proxyConfig.getProducerProcessorThreadPoolQueueCapacity()
        );

        this.serviceManager = serviceManager;

        this.producerProcessor = new ProducerProcessor(this, serviceManager, this.producerProcessorExecutor);
        //初始化客户端处理器，传入当前对象和服务管理器。
        //ClientProcessor客户端处理器负责处理客户端的请求和信息管理。
        this.clientProcessor = new ClientProcessor(this, serviceManager);
        this.init();
    }



    //初始化方法
    protected void init() {
        this.appendStartAndShutdown(this.serviceManager);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/30
     * @方法描述：创建集群模式下使用的默认处理器对象
     */
    public static DefaultMessagingProcessor createForClusterMode() {
        return createForClusterMode(null);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/30
     * @方法描述：创建集群模式下使用的默认处理器对象，这个方法可以设定RPC钩子函数
     */
    public static DefaultMessagingProcessor createForClusterMode(RPCHook rpcHook) {
        return new DefaultMessagingProcessor(ServiceManagerFactory.createForClusterMode(rpcHook));
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/30
     * @方法描述：创建本地模式下使用的默认处理器对象
     */
    public static DefaultMessagingProcessor createForLocalMode(BrokerController brokerController) {
        return createForLocalMode(brokerController, null);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/30
     * @方法描述：创建本地模式下使用的默认处理器对象，这个方法可以设定RPC钩子函数
     */
    public static DefaultMessagingProcessor createForLocalMode(BrokerController brokerController,RPCHook rpcHook) {
        return new DefaultMessagingProcessor(ServiceManagerFactory.createForLocalMode(brokerController, rpcHook));
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/30
     * @方法描述：根据主题获取路由信息的方法
     */
    @Override
    public ProxyTopicRouteData getTopicRouteDataForProxy(ProxyContext ctx, List<Address> requestHostAndPortList,
                                                         String topicName) throws Exception {
        //从服务管理器中获取主题路由服务，然后使用主题路由服务获取主题路由信息
        return this.serviceManager.getTopicRouteService().getTopicRouteForProxy(ctx, requestHostAndPortList, topicName);
    }


    //得到元数据服务组件的方法
    @Override
    public MetadataService getMetadataService() {
        return this.serviceManager.getMetadataService();
    }

    @Override
    public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, QueueSelector queueSelector, String producerGroup, int sysFlag, List<Message> msg, long timeoutMillis) {
        return this.producerProcessor.sendMessage(ctx, queueSelector, producerGroup, sysFlag, msg, timeoutMillis);
    }

    @Override
    public SubscriptionGroupConfig getSubscriptionGroupConfig(ProxyContext ctx, String consumerGroupName) {
        return this.serviceManager.getMetadataService().getSubscriptionGroupConfig(ctx, consumerGroupName);
    }


    @Override
    public CompletableFuture<RemotingCommand> request(ProxyContext ctx, String brokerName, RemotingCommand request, long timeoutMillis) {
        return null;
    }

    @Override
    public CompletableFuture<Void> requestOneway(ProxyContext ctx, String brokerName, RemotingCommand request, long timeoutMillis) {
        return null;
    }

    @Override
    public void registerProducer(ProxyContext ctx, String producerGroup, ClientChannelInfo clientChannelInfo) {
        this.clientProcessor.registerProducer(ctx, producerGroup, clientChannelInfo);
    }

    @Override
    public void unRegisterProducer(ProxyContext ctx, String producerGroup, ClientChannelInfo clientChannelInfo) {
        this.clientProcessor.unRegisterProducer(ctx, producerGroup, clientChannelInfo);
    }

    @Override
    public Channel findProducerChannel(ProxyContext ctx, String producerGroup, String clientId) {
        return this.clientProcessor.findProducerChannel(ctx, producerGroup, clientId);
    }

    @Override
    public void registerProducerListener(ProducerChangeListener producerChangeListener) {
        this.clientProcessor.registerProducerChangeListener(producerChangeListener);
    }

    @Override
    public void registerConsumer(ProxyContext ctx, String consumerGroup, ClientChannelInfo clientChannelInfo, ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere, Set<SubscriptionData> subList, boolean updateSubscription) {
        this.clientProcessor.registerConsumer(ctx, consumerGroup, clientChannelInfo, consumeType, messageModel, consumeFromWhere, subList, updateSubscription);
    }

    @Override
    public ClientChannelInfo findConsumerChannel(ProxyContext ctx, String consumerGroup, Channel channel) {
        return this.clientProcessor.findConsumerChannel(ctx, consumerGroup, channel);
    }

    @Override
    public void unRegisterConsumer(ProxyContext ctx, String consumerGroup, ClientChannelInfo clientChannelInfo) {
        this.clientProcessor.unRegisterConsumer(ctx, consumerGroup, clientChannelInfo);
    }

    @Override
    public void registerConsumerListener(ConsumerIdsChangeListener consumerIdsChangeListener) {
        this.clientProcessor.registerConsumerIdsChangeListener(consumerIdsChangeListener);
    }

    @Override
    public void doChannelCloseEvent(String remoteAddr, Channel channel) {
        this.clientProcessor.doChannelCloseEvent(remoteAddr, channel);
    }

    @Override
    public ConsumerGroupInfo getConsumerGroupInfo(ProxyContext ctx, String consumerGroup) {
        return this.clientProcessor.getConsumerGroupInfo(ctx, consumerGroup);
    }

    @Override
    public ProxyRelayService getProxyRelayService() {
        return this.serviceManager.getProxyRelayService();
    }


}
